package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.DwsTrafficVideoPlayBean;
import com.atguigu.edu.realtime.common.kafkaTopics;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.Map;

public class DwsTrafficVideoPlayWindow {

    public static void main(String[] args) throws Exception {
        // TODO: 2022/9/3 获取页面日志 根据 pageId = home 或者 course_detail 构建bean
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topic_name = kafkaTopics.DWD_TRAFFIC_VIDEO_LOG;
        String groupId = "dws_traffic_video_play_window";

        /**
         * {
         *   "appVideo": {             --视频信息
         *     "play_sec": 19,         --播放时长
         *     "position_sec":390,     --播放进度
         *     "video_id": "3904"      --视频id
         *   },
         *   "common": {
         *     "ar": "4",
         *     "ba": "Sumsung",
         *     "ch": "oppo",
         *     "is_new": "0",
         *     "md": "Sumsung Galaxy S20",
         *     "mid": "mid_253",
         *     "os": "Android 11.0",
         *     "sc": "1",
         *     "sid": "47157c4a-4790-4b9a-a859-f0d36cd62a10",
         *     "uid": "329",
         *     "vc": "v2.1.134"
         *   },
         *   "ts": 1645526307119
         * }
         */
        SingleOutputStreamOperator<JSONObject> map = env.addSource(KafkaUtil.getKafkaConsumer(topic_name, groupId))
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        //获取原始日志数据
                        JSONObject playLog = JSON.parseObject(value);

                        long ts = MapUtils.getLongValue(playLog, "ts");
                        JSONObject appVideo = playLog.getJSONObject("appVideo");
                        Long video_id = MapUtils.getLong(appVideo, "video_id");
                        Long play_sec = MapUtils.getLong(appVideo, "play_sec");
                        String session_id = playLog.getJSONObject("common").getString("sid");
                        String userId = playLog.getJSONObject("common").getString("uid");

                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("ts", ts);
                        jsonObject.put("video_id", video_id);
                        jsonObject.put("play_sec", play_sec);
                        jsonObject.put("session_id", session_id);
                        jsonObject.put("user_id", userId);
                        return jsonObject;
                    }
                });

        KeyedStream<JSONObject, Tuple2<String, String>> keyedStream = map
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return MapUtils.getLongValue(element, "ts");
                    }
                }))
                .keyBy(new KeySelector<JSONObject, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(JSONObject value) throws Exception {
                        return Tuple2.of(MapUtils.getString(value, "session_id"), MapUtils.getString(value, "video_id"));
                    }
                });


        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("begin")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        Long video_id = MapUtils.getLong(value, "video_id");
                        return video_id != null;
                    }
                })
                .within(Time.seconds(35)).next("end").where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value, Context<JSONObject> ctx) throws Exception {
                        Long video_id = MapUtils.getLong(value, "video_id");
                        return video_id != null;
                    }
                });

        OutputTag<JSONObject> later_data_out_tag = new OutputTag<JSONObject>("later_data", TypeInformation.of(JSONObject.class));

        SingleOutputStreamOperator<JSONObject> patternStream = CEP.<JSONObject>pattern(keyedStream, pattern).select(later_data_out_tag, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp) throws Exception {

                List<JSONObject> begin = pattern.get("begin");
                JSONObject jsonObject = begin.get(0);
                return jsonObject;
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> pattern) throws Exception {
                List<JSONObject> begin = pattern.get("begin");
                JSONObject jsonObject = begin.get(0);
                return jsonObject;
            }
        });


        DataStream<JSONObject> lateDataStream = patternStream.getSideOutput(later_data_out_tag);

        //得到最终的一次播放数据
        DataStream<JSONObject> unionStream = lateDataStream.union(patternStream);


//        unionStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessAllWindowFunction<JSONObject, Object, TimeWindow>() {
//
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//
//                ValueStateDescriptor<Map> valueStateDescriptor = new ValueStateDescriptor("ss",Map.class);
//                getRuntimeContext().getState(valueStateDescriptor);
//            }
//
//            @Override
//            public void process(ProcessAllWindowFunction<JSONObject, Object, TimeWindow>.Context context,
//                                Iterable<JSONObject> elements, Collector<Object> out) throws Exception {
//
//            }
//        });


        KeyedStream<JSONObject, String> keyedStream_user_id = unionStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return MapUtils.getString(value, "user_id");
            }

        });


        SingleOutputStreamOperator<DwsTrafficVideoPlayBean> outputStreamOperator = keyedStream_user_id.window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<JSONObject, DwsTrafficVideoPlayBean, String, TimeWindow>() {

                    ValueState<String> state = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("lastDt", String.class);
                        state = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void process(String userId,
                                        ProcessWindowFunction<JSONObject, DwsTrafficVideoPlayBean, String, TimeWindow>.Context context,
                                        Iterable<JSONObject> elements, Collector<DwsTrafficVideoPlayBean> out) throws Exception {

                        TimeWindow window = context.window();
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());


                        elements.forEach(item -> {
                            Long ts = MapUtils.getLong(item, "ts");
                            String date = DateFormatUtil.toDate(ts);

                            try {
                                String value = state.value();

//                                某个用户当天首次播放 记录一次 setUser_cnt
                                if (StringUtils.isBlank(value) || !value.equals(date)) {
                                    state.update(date);

                                    DwsTrafficVideoPlayBean resultBean = generateResBean(stt, edt, item);
                                    resultBean.setUser_cnt(1L);
                                    resultBean.setTs(System.currentTimeMillis());

                                    out.collect(resultBean);
                                }
//                                非首次播放  只记录播放次数setPlay_cnt 但是 setUser_cnt为0
                                else {
                                    DwsTrafficVideoPlayBean resultBean = generateResBean(stt, edt, item);
                                    resultBean.setUser_cnt(0L);
                                    resultBean.setTs(System.currentTimeMillis());
                                    out.collect(resultBean);
                                }

                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        });
                    }
                });


//(stt=2022-09-05 20:38:35, edt=2022-09-05 20:38:40, video_id=1747, user_cnt=0, during_time_sum=22, play_cnt=1, ts=1662425257293)
//(stt=2022-09-05 20:38:35, edt=2022-09-05 20:38:40, video_id=1747, user_cnt=0, during_time_sum=8, play_cnt=1, ts=1662425257293)

        String tableName = "dws_traffic_video_play_window";
        SinkFunction<DwsTrafficVideoPlayBean> sinkFunc = ClickHouseUtil.getSinkFunc(tableName, DwsTrafficVideoPlayBean.class);

        outputStreamOperator.addSink(sinkFunc);

        env.execute();

    }

    private static DwsTrafficVideoPlayBean generateResBean(String stt, String edt, JSONObject item) {
        DwsTrafficVideoPlayBean resultBean = new DwsTrafficVideoPlayBean();
        resultBean.setPlay_cnt(1L);
        resultBean.setVideo_id(MapUtils.getLong(item, "video_id"));
        resultBean.setVideo_name("");
        resultBean.setStt(stt);
        resultBean.setEdt(edt);
        resultBean.setDuring_time_sum(MapUtils.getLong(item, "play_sec"));

        return resultBean;
    }
}
